package blue.mqtt4j.internal.client.impl;

import blue.mqtt4j.client.ConsumerListener;
import blue.mqtt4j.client.MqttClientException;
import blue.mqtt4j.client.MqttMessage;
import blue.mqtt4j.client.MqttTopic;
import blue.mqtt4j.client.codec.Decoder;
import blue.mqtt4j.core.topic.MqttTopicUtil;
import blue.mqtt4j.internal.client.net.NettyClient;
import blue.mqtt4j.internal.client.task.TaskExecutor;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author Jin Zheng
 * @since 1.0 2020-10-16
 */
public class ReceiveHandler
{
	private static Logger logger = LoggerFactory.getLogger(ReceiveHandler.class);

	private final ConcurrentMap<MqttTopic, ConsumerListener> listenerMap = new ConcurrentHashMap<>();
	private final ConcurrentMap<MqttTopic, Decoder> decoderMap = new ConcurrentHashMap<>();
	private final ConcurrentMap<Integer, Topic> topicMap = new ConcurrentHashMap<>();
	private final ConcurrentMap<Integer, MqttMessage> messageMap = new ConcurrentHashMap<>();
	private final TaskExecutor executor;
	private final NettyClient client;

	public ReceiveHandler(TaskExecutor executor, NettyClient client)
	{
		this.executor = executor;
		this.client = client;
	}

	public void add(List<MqttTopic> topicList, Decoder decoder, ConsumerListener listener)
	{
		for (MqttTopic topic : topicList)
		{
			if (listenerMap.containsKey(topic))
			{
				throw new MqttClientException(topic + " exits");
			}
			listenerMap.put(topic, listener);
			if (decoder != null)
			{
				decoderMap.put(topic, decoder);
			}
		}
	}

	public void addUnsubscribeTopic(int messageId, Topic topic)
	{
		topicMap.put(messageId, topic);
	}

	public void remove(int messageId)
	{
		Topic topic = topicMap.get(messageId);
		if (topic == null)
		{
			return;
		}
		listenerMap.remove(topic);
	}

	public boolean check(Topic topic)
	{
		return listenerMap.containsKey(topic);
	}

	public void receive(MqttPublishMessage mqttMessage)
	{
		MqttMessage message = new DefaultMqttMessage(mqttMessage);
		MqttQoS qos = mqttMessage.fixedHeader().qosLevel();
		int messageId = mqttMessage.variableHeader().packetId();
		switch (qos)
		{
			case AT_MOST_ONCE:
				this.receive(message);
				break;
			case AT_LEAST_ONCE:
				this.receive(message);
				client.sendPubAck(messageId);
				break;
			case EXACTLY_ONCE:
				messageMap.put(messageId, message);
				client.sendPubRec(messageId);
				break;
			default:
				break;
		}
	}

	public void pubRel(int messageId)
	{
		MqttMessage message = messageMap.get(messageId);
		if (message != null)
		{
			this.receive(message);
			messageMap.remove(messageId);
		}

		client.sendPubComp(messageId);
	}

	public Set<MqttTopic> topicSet()
	{
		return listenerMap.keySet();
	}

	private void receive(MqttMessage message)
	{
		for (Map.Entry<MqttTopic, ConsumerListener> entry : listenerMap.entrySet())
		{
			if (MqttTopicUtil.isMatched(entry.getKey().topic(), message.topic()))
			{
				Decoder decoder = decoderMap.get(entry.getKey());
				executor.execute(() ->
				{
					try
					{
						if (decoder == null)
						{
							entry.getValue().onReceived(message);
						}
						else
						{
							Object object = decoder.decode((byte[])message.payload());
							MqttMessage decodeMessage = new DefaultMqttMessage(message.topic(), message.qos(), object);
							entry.getValue().onReceived(decodeMessage);
						}
					}
					catch (Exception e)
					{
						logger.error("Error, ", e);
					}
				});
			}
		}
	}

}
